package com.example.demo.basic;


import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @ClassName SpringProducer
 * @Description
 * @Author zhangdachen
 * @Date 2021-09-14 9:35
 * @Version V1.0
 **/
@Component
public class SpringProducer {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * @description: 发送普通消息
     * @param: [topic, msg]
     * @return: void
     * @author: ZhangDaChen
     * @date: 2021-09-14
     */
    public void sendMessage(String topic, String msg) {
        this.rocketMQTemplate.convertAndSend(topic, msg);
    }

    public void sendMessageInTransaction(String topic, String msg) throws InterruptedException {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            //尝试在head中加入自定义属性
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader(RocketMQHeaders.TRANSACTION_ID, "TransID_" + i)
                    //发送到事务监听器之后，自己设定的TAGS会丢失。但上面那个属性会丢失。
                    .setHeader(RocketMQHeaders.TAGS, tags[i % tags.length])
                    //MyProp
                    .setHeader("MyProp", "MyProp_" + i)
                    .build();
            String destination = topic + ":" + tags[i % tags.length];
            //发送事务消息时，转化成RocketMq的Message对象，在调用rockermq的api完成事务消息机制
            SendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
            System.out.printf("%s%n", transactionSendResult);
            Thread.sleep(10);

        }
    }
}
